package com.orion.controller;

import cn.hutool.json.JSONUtil;
import com.orion.entity.TestMsg;
import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author Administrator
 * @date 2021/12/15
 */
@RestController
@RequestMapping("/transMsg")
public class TransactionMsgController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 同步发送
     *
     * @param topic
     * @param msg
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @GetMapping("/sendSimpleSync")
    public void sendSimpleSync(String topic, String key, TestMsg msg) throws ExecutionException, InterruptedException, TimeoutException {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @SneakyThrows
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {
                String jsonUser = JSONUtil.toJsonStr(msg);
                ListenableFuture<SendResult<String, Object>> future = kafkaOperations.send(topic, key, jsonUser);
                SendResult<String, Object> sendResult = future.get(10, TimeUnit.SECONDS);
                System.out.println(sendResult);
                throw new RuntimeException("fail");
                //return true;
            }
        });

    }

    @GetMapping("/sendSimpleSingle")
    public void sendSimple(String topic, String key, TestMsg msg) {
        String jsonUser = JSONUtil.toJsonStr(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, jsonUser);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("send msg fail:" + ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("send msg success:" + result.toString());
            }
        });
        throw new RuntimeException("fail");
    }


    @GetMapping("/sendSimpleMulti")
    public void sendSimple(String topic1,String topic2, String key, TestMsg msg) {
        String jsonUser = JSONUtil.toJsonStr(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic1, key, jsonUser);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("send msg fail1:" + ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("send msg success1:" + result.toString());
            }
        });
        ListenableFuture<SendResult<String, Object>> future2 = kafkaTemplate.send(topic2, key, jsonUser);
        future2.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("send msg fail2:" + ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("send msg success2:" + result.toString());
            }
        });
        throw new RuntimeException("fail");
    }


    @GetMapping("/sendMoreSimpleSingle")
    public void sendMoreSimple(String topic1, Integer times, String key, String sign, TestMsg msg) {
        for (int i = 0; i < times; i++) {
            msg.setName(sign + " : " + i);
            String jsonUser = JSONUtil.toJsonStr(msg);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic1, (key != null ? key + i : null), jsonUser);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("send msg fail:" + ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("send msg success:" + result.toString());
                }
            });
        }
        throw new RuntimeException("fail");
    }


    @GetMapping("/sendMoreSimpleMulti")
    public void sendMoreSimple(String topic1,String topic2, Integer times, String key, String sign, TestMsg msg) {
        for (int i = 0; i < times; i++) {
            msg.setName(sign + " : " + i);
            String jsonUser = JSONUtil.toJsonStr(msg);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic1, (key != null ? key + i : null), jsonUser);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("send msg fail1:" + ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("send msg success1:" + result.toString());
                }
            });

            ListenableFuture<SendResult<String, Object>> future2 = kafkaTemplate.send(topic2, (key != null ? key + i : null), jsonUser);
            future2.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("send msg fail2:" + ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("send msg success2:" + result.toString());
                }
            });

        }
        throw new RuntimeException("fail");
    }
}
